package com.example.klinemqflink.config;

import com.example.klinemqflink.model.MarketData;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.StateBackendOptions;
import org.apache.flink.contrib.streaming.state.RocksDBNativeMetricOptions;
import org.apache.flink.contrib.streaming.state.RocksDBOptions;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class FlinkConfig{

    @Bean
    public StreamExecutionEnvironment flinkEnv() throws Exception {

        // 配置 RocksDB 状态后端
//        String checkpointPath = "/Users/zhangrui/flink_cache"; // 替换为实际的检查点路径
        // 1. 启用 RocksDB 状态后端（减少内存压力）
//        org.apache.flink.configuration.Configuration config = new org.apache.flink.configuration.Configuration();
//        config.setString(StateBackendOptions.STATE_BACKEND, "rocksdb");
//        config.setString(RocksDBOptions.LOCAL_DIRECTORIES, checkpointPath); // 指定本地存储路径
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(12);
        env.enableCheckpointing(60_000); // 启用检查点

        // 2. 配置重启策略
        env.setRestartStrategy(
                RestartStrategies.fixedDelayRestart(
                        3,      // 最大重启尝试次数（含首次运行）
                        10000   // 重启间隔（毫秒）
                )
        );
        return env;
    }

//    @Bean
//    public DataStream<MarketData> marketDataStream(StreamExecutionEnvironment flinkEnv) {
//        return flinkEnv.fromData(new MarketData());
//    }
}
